package com.zcy;

import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.runtime.client.EPEventService;
import com.zcy.event.MyEvent;
import com.zcy.util.EsperBuilder;
import com.zcy.util.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

/**
 * 14.3.1 Length Window (length or win:length)
 * 称为滑动窗口（Sliding Window)
 */
@Slf4j
public class LengthWindow extends TimeUtils {

    private static final String _Length = "@name('length-statement') select sum(num) from MyEvent#win:length(2)";
    private static final String _GroupWin = "@name('group-statement') select sum(num) from MyEvent#groupwin(amount)#win:length(2) group by amount";
    private static final String _LastEvent = "@name('last-statement') select sum(num) from MyEvent#lastevent";

    /**
     * 普通Length
     * sum is : 10
     * sum is : 15
     * sum is : 25
     * sum is : 50
     * sum is : 70
     */
    @Test
    public void lengthWindowTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_Length, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("sum is : {}", eventBean.get("sum(num)"));
            }
        });

        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 1L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(5).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 1L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(20).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 3L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("003").name("003-name").num(30).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 4L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("004").name("004-name").num(40).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }

    /**
     * 分组Length
     * sum is : 10
     * sum is : 30
     * sum is : 50
     * sum is : 20
     * sum is : 30
     * sum is : 40
     */
    @Test
    public void groupWinTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_GroupWin, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("sum is : {}", eventBean.get("sum(num)"));
            }
        });

        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 2L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(20).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 2L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(30).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 2L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(20).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 3L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("003").name("003-name").num(30).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 4L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("004").name("004-name").num(40).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }

    /**
     * LastEvent = length(1)
     * sum is : 10
     * sum is : 20
     * sum is : 30
     * sum is : 20
     * sum is : 30
     * sum is : 40
     */
    @Test
    public void lastEventTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_LastEvent, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("sum is : {}", eventBean.get("sum(num)"));
            }
        });

        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 2L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(20).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 2L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(30).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 2L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(20).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 3L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("003").name("003-name").num(30).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 4L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("004").name("004-name").num(40).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }

}
